package cn.doitedu.dao;

import cn.doitedu.pojo.CombineCondition;
import cn.doitedu.pojo.EventBean;
import cn.doitedu.pojo.EventParam;
import cn.doitedu.pojo.LogBean;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.java.utils.ParameterTool;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 优点：ClickHouse擅长单表查询，擅长作为OLAP引擎，多条件即席查询
 * 缺点：不支持join，并发能力差
 */
public class HistoryEventClickHouseDaoImpl implements HistoryEventDao {

    private Connection connection;

    @Override
    public void init(ParameterTool parameterTool, MapState<String, String> cacheState) throws Exception {

        connection = DriverManager.getConnection(parameterTool.getRequired("clickhouse.url"));

    }

    @Override
    public String queryEventStr(LogBean bean, CombineCondition combineCondition) throws Exception {
        //查询ClickHouse，并且将行为字符串返回
        StringBuffer sb = new StringBuffer();
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        try {
            preparedStatement = connection.prepareStatement(combineCondition.getSql());
            preparedStatement.setString(1, bean.getDeviceId());
            preparedStatement.setLong(2, combineCondition.getStartTime());
            preparedStatement.setLong(3, combineCondition.getEndTime());
            //返回的resultSet中，装的是多个行为
            resultSet = preparedStatement.executeQuery();
            //用来装实际从clickhouse中查询出的事件（各种各样的事件，可能很多个）
            ArrayList<String> events = new ArrayList<>();
            while (resultSet.next()) {
                String eventStr = resultSet.getString(1);
                events.add(eventStr);
            }
            //我实际关系的即要匹配的事件
            List<String> targetEvents = combineCondition.getEvents().stream().map(EventParam::getEventId).collect(Collectors.toList());
            //将实际查出来的事件与要匹配的事件完成映射
            for (String event : events) {
                int index = targetEvents.indexOf(event);
                sb.append(index+1);
            }
            //完成正则匹配的功能
        } finally {
            if (resultSet != null) {
                resultSet.close();
            }
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
        return sb.toString();
    }

    @Override
    public void close() throws Exception {

        connection.close();
    }
}
